package com.jumei.flume.handler.shuabao;

import com.jumei.flume.bo.ShuabaoSensorMessage;
import com.jumei.flume.filter.EventsFilter;
import com.jumei.flume.handler.Handler;
import com.jumei.flume.parser.JSONStringToShuabaoSensorMessageConverter;
import com.jumei.flume.parser.JsonSplit2HdfsFormatter4Shuabao;
import org.apache.flume.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;

public class ShuabaoSensorKafkaHandler implements Handler {
	private final static Logger LOGGER = LoggerFactory.getLogger(ShuabaoSensorKafkaHandler.class);

	JsonSplit2HdfsFormatter4Shuabao hdfsFormatter;
	static JSONStringToShuabaoSensorMessageConverter converter = new JSONStringToShuabaoSensorMessageConverter();
	
	@Override
	public void start() {
		hdfsFormatter = new JsonSplit2HdfsFormatter4Shuabao();
		//kafka初始化
	}

	@Override
	public void stop() {}

	@Override
	public void handler(List<Event> eventList) throws Exception {
		
		if(CollectionUtils.isEmpty(eventList)) {
			return;
		}

		for(Event event : eventList) {

			try {
				String line = new String(event.getBody());
				ShuabaoSensorMessage sensorMessage = converter.convert(line);
	            if (null == sensorMessage) {
	                continue;
	            }
	            
	            //只选择指定项目的指定事件
	            if (!EventsFilter.doFilter4Shuabao(sensorMessage)) {
	            	continue;
	            }
				
				
				String hdfsLine = hdfsFormatter.convert2HdfsFormatter4Asm(sensorMessage);
				event.setBody(hdfsLine.getBytes());
				
			} catch (Exception e) {
				//say something here
				LOGGER.error("handler error: ", e);
			}
			
		}
		
		LOGGER.info("Get kafka records: {}.", eventList.size());
	}

}
